package org.weishe.rtr.action;

import java.util.Calendar;

import javax.servlet.http.HttpServletRequest;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.weishe.rtr.mr.Step1;
import org.weishe.rtr.mr.Step1.S1Mapper;
import org.weishe.util.Util;

@Controller
@RequestMapping(value = "/mapreduceAction")
public class MapreduceAction {

	Calendar c = Calendar.getInstance();

	@RequestMapping(value = "/run")
	@ResponseBody
	public String run(HttpServletRequest request) {
		System.out.println("开始执行mapreduce程序");
		runStep1();
		return "success";
	}

	public static void main(String[] args) {
		MapreduceAction a = new MapreduceAction();
		a.runStep1();
	}

	private void runStep1() {
		Configuration config = new Configuration();
		// config.set("fs.defaultFS", "hdfs://node2:8020");
		// config.set("yarn.resourcemanager.hostname", "node3");
		// config.set("mapred.jar", Util.getMrLibPath() + "mapreduce.jar");
		config.set("mapred.jar", "/Users/chenbiao/Documents/workspace/rtr/WebContent/WEB-INF/mrlib/mapreduce.jar");
		//
		try {
			FileSystem fs = FileSystem.get(config);
			Job job = Job.getInstance(config);
			job.setJarByClass(Step1.class);

			job.setJobName("1.数据清洗");

			job.setMapperClass(S1Mapper.class);

			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(NullWritable.class);

			// 设置MR执行的输入文件
			FileInputFormat.addInputPath(job, new Path("/rtr/*"));

			// 该目录表示MR执行之后的结果数据所在目录，必须不能存在
			Path outputPath = new Path("/rtrout/" + c.getTimeInMillis());

			FileOutputFormat.setOutputPath(job, outputPath);

			boolean f = job.waitForCompletion(true);
			if (f) {
				System.out.println("step1 成功执行");
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
